package com.swak.registry;

import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

import com.swak.Constants;
import com.swak.reactivex.threads.SwakThreadFactory;
import com.swak.registry.retry.FailedNotifiedTask;
import com.swak.registry.retry.FailedRegisteredTask;
import com.swak.registry.retry.FailedSubscribedTask;
import com.swak.registry.retry.FailedUnregisteredTask;
import com.swak.registry.retry.FailedUnsubscribedTask;
import com.swak.timer.HashedWheelTimer;
import com.swak.utils.CollectionUtils;

public abstract class FailbackRegistry extends AbstractRegistry {

	private final ConcurrentMap<URL, FailedRegisteredTask> failedRegistered = new ConcurrentHashMap<URL, FailedRegisteredTask>();

	private final ConcurrentMap<URL, FailedUnregisteredTask> failedUnregistered = new ConcurrentHashMap<URL, FailedUnregisteredTask>();

	private final ConcurrentMap<Holder, FailedSubscribedTask> failedSubscribed = new ConcurrentHashMap<Holder, FailedSubscribedTask>();

	private final ConcurrentMap<Holder, FailedUnsubscribedTask> failedUnsubscribed = new ConcurrentHashMap<Holder, FailedUnsubscribedTask>();

	private final ConcurrentMap<Holder, FailedNotifiedTask> failedNotified = new ConcurrentHashMap<Holder, FailedNotifiedTask>();

	/**
	 * The time in milliseconds the retryExecutor will wait
	 */
	private final int retryPeriod;

	// Timer for failure retry, regular check if there is a request for failure, and
	// if there is, an unlimited retry
	private final HashedWheelTimer retryTimer;

	public FailbackRegistry(URL url) {
		super(url);

		this.retryPeriod = url.getParameter(Constants.REGISTRY_RETRY_PERIOD_KEY,
				Constants.DEFAULT_REGISTRY_RETRY_PERIOD);

		// since the retry task will not be very much. 128 ticks is enough.
		retryTimer = new HashedWheelTimer(new SwakThreadFactory("Registry.RetryTimer-", true, new AtomicInteger(0)),
				retryPeriod, TimeUnit.MILLISECONDS, 128);
	}

	public void removeFailedRegisteredTask(URL url) {
		failedRegistered.remove(url);
	}

	public void removeFailedUnregisteredTask(URL url) {
		failedUnregistered.remove(url);
	}

	public void removeFailedSubscribedTask(URL url, NotifyListener listener) {
		Holder h = new Holder(url, listener);
		failedSubscribed.remove(h);
	}

	public void removeFailedUnsubscribedTask(URL url, NotifyListener listener) {
		Holder h = new Holder(url, listener);
		failedUnsubscribed.remove(h);
	}

	public void removeFailedNotifiedTask(URL url, NotifyListener listener) {
		Holder h = new Holder(url, listener);
		failedNotified.remove(h);
	}

	private void addFailedRegistered(URL url) {
		FailedRegisteredTask oldOne = failedRegistered.get(url);
		if (oldOne != null) {
			return;
		}
		FailedRegisteredTask newTask = new FailedRegisteredTask(url, this);
		oldOne = failedRegistered.putIfAbsent(url, newTask);
		if (oldOne == null) {
			// never has a retry task. then start a new task for retry.
			retryTimer.newTimeout(newTask, retryPeriod, TimeUnit.MILLISECONDS);
		}
	}

	private void removeFailedRegistered(URL url) {
		FailedRegisteredTask f = failedRegistered.remove(url);
		if (f != null) {
			f.cancel();
		}
	}

	private void addFailedUnregistered(URL url) {
		FailedUnregisteredTask oldOne = failedUnregistered.get(url);
		if (oldOne != null) {
			return;
		}
		FailedUnregisteredTask newTask = new FailedUnregisteredTask(url, this);
		oldOne = failedUnregistered.putIfAbsent(url, newTask);
		if (oldOne == null) {
			// never has a retry task. then start a new task for retry.
			retryTimer.newTimeout(newTask, retryPeriod, TimeUnit.MILLISECONDS);
		}
	}

	private void removeFailedUnregistered(URL url) {
		FailedUnregisteredTask f = failedUnregistered.remove(url);
		if (f != null) {
			f.cancel();
		}
	}

	protected void addFailedSubscribed(URL url, NotifyListener listener) {
		Holder h = new Holder(url, listener);
		FailedSubscribedTask oldOne = failedSubscribed.get(h);
		if (oldOne != null) {
			return;
		}
		FailedSubscribedTask newTask = new FailedSubscribedTask(url, this, listener);
		oldOne = failedSubscribed.putIfAbsent(h, newTask);
		if (oldOne == null) {
			// never has a retry task. then start a new task for retry.
			retryTimer.newTimeout(newTask, retryPeriod, TimeUnit.MILLISECONDS);
		}
	}

	private void removeFailedSubscribed(URL url, NotifyListener listener) {
		Holder h = new Holder(url, listener);
		FailedSubscribedTask f = failedSubscribed.remove(h);
		if (f != null) {
			f.cancel();
		}
		removeFailedUnsubscribed(url, listener);
		removeFailedNotified(url, listener);
	}

	private void addFailedUnsubscribed(URL url, NotifyListener listener) {
		Holder h = new Holder(url, listener);
		FailedUnsubscribedTask oldOne = failedUnsubscribed.get(h);
		if (oldOne != null) {
			return;
		}
		FailedUnsubscribedTask newTask = new FailedUnsubscribedTask(url, this, listener);
		oldOne = failedUnsubscribed.putIfAbsent(h, newTask);
		if (oldOne == null) {
			// never has a retry task. then start a new task for retry.
			retryTimer.newTimeout(newTask, retryPeriod, TimeUnit.MILLISECONDS);
		}
	}

	private void removeFailedUnsubscribed(URL url, NotifyListener listener) {
		Holder h = new Holder(url, listener);
		FailedUnsubscribedTask f = failedUnsubscribed.remove(h);
		if (f != null) {
			f.cancel();
		}
	}

	private void addFailedNotified(URL url, NotifyListener listener, List<URL> urls) {
		Holder h = new Holder(url, listener);
		FailedNotifiedTask newTask = new FailedNotifiedTask(url, listener);
		FailedNotifiedTask f = failedNotified.putIfAbsent(h, newTask);
		if (f == null) {
			// never has a retry task. then start a new task for retry.
			newTask.addUrlToRetry(urls);
			retryTimer.newTimeout(newTask, retryPeriod, TimeUnit.MILLISECONDS);
		} else {
			// just add urls which needs retry.
			newTask.addUrlToRetry(urls);
		}
	}

	private void removeFailedNotified(URL url, NotifyListener listener) {
		Holder h = new Holder(url, listener);
		FailedNotifiedTask f = failedNotified.remove(h);
		if (f != null) {
			f.cancel();
		}
	}

	ConcurrentMap<URL, FailedRegisteredTask> getFailedRegistered() {
		return failedRegistered;
	}

	ConcurrentMap<URL, FailedUnregisteredTask> getFailedUnregistered() {
		return failedUnregistered;
	}

	ConcurrentMap<Holder, FailedSubscribedTask> getFailedSubscribed() {
		return failedSubscribed;
	}

	ConcurrentMap<Holder, FailedUnsubscribedTask> getFailedUnsubscribed() {
		return failedUnsubscribed;
	}

	ConcurrentMap<Holder, FailedNotifiedTask> getFailedNotified() {
		return failedNotified;
	}

	@Override
	public void register(URL url) {
		super.register(url);
		removeFailedRegistered(url);
		removeFailedUnregistered(url);
		try {
			// Sending a registration request to the server side
			doRegister(url);
		} catch (Exception e) {
			Throwable t = e;
			logger.error("Failed to register " + url + ", waiting for retry, cause: " + t.getMessage(), t);
			// Record a failed registration request to a failed list, retry regularly
			addFailedRegistered(url);
		}
	}

	@Override
	public void unregister(URL url) {
		super.unregister(url);
		removeFailedRegistered(url);
		removeFailedUnregistered(url);
		try {
			// Sending a cancellation request to the server side
			doUnregister(url);
		} catch (Exception e) {
			Throwable t = e;
			logger.error("Failed to unregister " + url + ", waiting for retry, cause: " + t.getMessage(), t);
			// Record a failed registration request to a failed list, retry regularly
			addFailedUnregistered(url);
		}
	}

	@Override
	public void subscribe(URL url, NotifyListener listener) {
		super.subscribe(url, listener);
		removeFailedSubscribed(url, listener);
		try {
			// Sending a subscription request to the server side
			doSubscribe(url, listener);
		} catch (Exception e) {
			Throwable t = e;

			List<URL> urls = getCacheUrls(url);
			if (CollectionUtils.isNotEmpty(urls)) {
				notify(url, listener, urls);
				logger.error("Failed to subscribe " + url + ", Using cached list: " + urls + " from cache file: "
						+ this.getCacheFile().getName() + ", cause: " + t.getMessage(), t);
			} else {
				logger.error("Failed to subscribe " + url + ", waiting for retry, cause: " + t.getMessage(), t);
			}

			// Record a failed registration request to a failed list, retry regularly
			addFailedSubscribed(url, listener);
		}
	}

	@Override
	public void unsubscribe(URL url, NotifyListener listener) {
		super.unsubscribe(url, listener);
		removeFailedSubscribed(url, listener);
		try {
			// Sending a canceling subscription request to the server side
			doUnsubscribe(url, listener);
		} catch (Exception e) {
			Throwable t = e;

			logger.error("Failed to unsubscribe " + url + ", waiting for retry, cause: " + t.getMessage(), t);

			// Record a failed registration request to a failed list, retry regularly
			addFailedUnsubscribed(url, listener);
		}
	}

	@Override
	protected void notify(URL url, NotifyListener listener, List<URL> urls) {
		if (url == null) {
			throw new IllegalArgumentException("notify url == null");
		}
		if (listener == null) {
			throw new IllegalArgumentException("notify listener == null");
		}
		try {
			doNotify(url, listener, urls);
		} catch (Exception t) {
			// Record a failed registration request to a failed list, retry regularly
			addFailedNotified(url, listener, urls);
			logger.error("Failed to notify for subscribe " + url + ", waiting for retry, cause: " + t.getMessage(), t);
		}
	}

	protected void doNotify(URL url, NotifyListener listener, List<URL> urls) {
		super.notify(url, listener, urls);
	}

	@Override
	protected void recover() throws Exception {
		// register
		Set<URL> recoverRegistered = new HashSet<URL>(getRegistered());
		if (!recoverRegistered.isEmpty()) {
			if (logger.isInfoEnabled()) {
				logger.info("Recover register url " + recoverRegistered);
			}
			for (URL url : recoverRegistered) {
				addFailedRegistered(url);
			}
		}
		// subscribe
		Map<URL, Set<NotifyListener>> recoverSubscribed = new HashMap<URL, Set<NotifyListener>>(getSubscribed());
		if (!recoverSubscribed.isEmpty()) {
			if (logger.isInfoEnabled()) {
				logger.info("Recover subscribe url " + recoverSubscribed.keySet());
			}
			for (Map.Entry<URL, Set<NotifyListener>> entry : recoverSubscribed.entrySet()) {
				URL url = entry.getKey();
				for (NotifyListener listener : entry.getValue()) {
					addFailedSubscribed(url, listener);
				}
			}
		}
	}

	@Override
	public void destroy() {
		try {
			super.destroy();
			retryTimer.stop();
		} catch (Exception e) {
		}
	}

	// ==== Template method ====

	public abstract void doRegister(URL url);

	public abstract void doUnregister(URL url);

	public abstract void doSubscribe(URL url, NotifyListener listener);

	public abstract void doUnsubscribe(URL url, NotifyListener listener);

	static class Holder {

		private final URL url;

		private final NotifyListener notifyListener;

		Holder(URL url, NotifyListener notifyListener) {
			if (url == null || notifyListener == null) {
				throw new IllegalArgumentException();
			}
			this.url = url;
			this.notifyListener = notifyListener;
		}

		@Override
		public int hashCode() {
			return url.hashCode() + notifyListener.hashCode();
		}

		@Override
		public boolean equals(Object obj) {
			if (obj instanceof Holder) {
				Holder h = (Holder) obj;
				return this.url.equals(h.url) && this.notifyListener.equals(h.notifyListener);
			} else {
				return false;
			}
		}
	}
}
